package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

object Demo2SourceFunction {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    env.setParallelism(2)

    /**
      * 自定义source
      * 实现SourceFunction接口
      * 实现里面的run方法----> 在run方法中获取数据发送到下游
      *
      *
      */

    val myDS: DataStream[Int] = env.addSource(new MySourceFunction)

    myDS.print()

    env.execute()

  }
}

/**
  *
  * ParallelSourceFunction   : 并行的source, 可以多个线程一个读取数据，需要解决数据重复问题
  * RichParallelSourceFunction: 并行的source，多了open和close 方法
  * RichSourceFunction ：单一的source，多了open和close 方法
  * SourceFunction ： 单一的source
  */

class MySourceFunction extends SourceFunction[Int] {

  /**
    * 在run 方法中获取数据，将数据发送到下游
    *
    * 可以在run  方法中链接外部的数据源获取数据
    *
    * @param ctx flink source的上下文对象，用于将数据发送到哦下游
    */

  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {

    var i = 0
    while (i < 10) {

      //将数据发送到下游
      ctx.collect(i)

      i += 1

      Thread.sleep(1000)
    }

  }

  /**
    * 任务在取消的时候调用，用于回收资源（关闭jdbc 链接）
    *
    */

  override def cancel(): Unit = {

  }
}


